package cloud.pangui.core.worker;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.TimeUnit;

/**
 * @Auther hepangui
 * @Date 2020/11/7 0007
 */
public abstract class Worker implements Runnable {
	/**
	 * logger
	 */
	private static Logger logger = LoggerFactory.getLogger(Worker.class);

	/**
	 * 首次创建时的message，如果没有这个，work线程和主线程的addMessage无法保证先后顺序，不能保证消息消费
	 */
	protected Object firstMessage;

	/**
	 * 创造worker的manager
	 */
	protected WorkerManager workerManager;

	/**
	 * 是否保持长期存在，用于维持线程池最小数量
	 */
	protected boolean keepAlive;

	/**
	 * 处理结果
	 *
	 * @param message
	 */
	public abstract void work(Object message);

	/**
	 * 当前线程
	 */
	protected Thread currentThread;

	/**
	 * 在工作中
	 */
	protected volatile boolean working = false;

	public void start() {
		Thread thread = new Thread(this);
		thread.setName("worker_" + this.getClass().getName() + "_" + System.currentTimeMillis());
		this.currentThread = thread;
		thread.start();
	}

	/**
	 * 从队列中获取消息，并执行
	 * 工作逻辑：
	 * 第一次创建出来时，从firstMessage中获取消息
	 * 之后，从syncQueue中等待workerManager放消息
	 * 获取到消息之后工作，在工作过程中，有可能因为所有work都忙
	 * messages队列中堆积了一定量的消息，此时，会从此队列中取
	 * 如果messages队列中没有消息，则回到syncQueue中等待
	 */
	@Override
	public void run() {
		Object message = null;
		while (true) {
			try {
				if (firstMessage != null) {
					message = firstMessage;
					firstMessage = null;
				}

				if (message == null) {
					//message为null，说明为闲时，忙队列中没有消息，此时监控syncQueue
					//如果为keepAlive状态，会一直阻塞等待，
					//否则，等待一段时间后销毁
					message = takeMessage();
				}

				if (message == null) {
					//此时获取不到消息，说明已经闲了很久了
					workerManager.workerCount.decrementAndGet();
					workerManager.removeDead(this);
					return;
				}

				working = true;
				work(message);
				//worker在忙时，有可能worker满了，消息堆积在message队列中而不是syncQueue中
				//忙完了之后，看队列中是否有，有就使用该队列进行work
				if (currentThread.isInterrupted()) {
					throw new InterruptedException();
				}
				message = workerManager.getMessages().poll();
			} catch (InterruptedException e) {
				logger.debug(Thread.currentThread().getName() + " interrupted");
				workerManager.workerCount.decrementAndGet();
				workerManager.removeDead(this);
				return;
			} catch (Exception e) {
				message = null;
				logger.error(e.getMessage(), e);
			} catch (Throwable e) {
				message = null;
			} finally {
				working = false;
			}
		}
	}

	/**
	 * 从manager的syncWorkMessage队列中获取消息
	 * work在闲时只会从syncWorkMessage中获取消息
	 *
	 * @return
	 */
	public Object takeMessage() throws InterruptedException {
		if (this.keepAlive) {
			return workerManager.getSyncWorkMessage().take();
		} else {
			return workerManager.getSyncWorkMessage().poll(workerManager.getMaxIdle(), TimeUnit.MILLISECONDS);
		}

	}

	/**
	 * 获取当前线程
	 */
	public Thread getCurrentThread() {
		return currentThread;
	}


	public void destory() {
		if(this.getCurrentThread() !=null){
			this.getCurrentThread().interrupt();
		}
		this.working =false;
	}

	public boolean isWorking(){
		return this.working;
	}
}
